package com.hq.tomato.ms.strategy;

import com.mongodb.client.result.UpdateResult;
import com.rabbitmq.client.Channel;
import com.hq.tomato.ms.config.TomatoMsProperties;
import com.hq.tomato.ms.entity.RetryMessage;
import com.hq.tomato.ms.util.JacksonUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;
import org.springframework.util.CollectionUtils;

import java.nio.charset.StandardCharsets;
import java.time.LocalDateTime;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;

import static com.hq.tomato.ms.constants.RetryMessageConstants.*;

/**
 *
 *
 * @Description: <br>
 * @Project: <br>
 * @CreateDate: Created in 2019-11-18 20:36 <br>
 * @Author: <a href="1026365768@qq.com">qiang.hu</a>
 */
@Slf4j
@EnableConfigurationProperties(TomatoMsProperties.class)
public class MongoDbDlqStrategy implements DlqStrategy {
    /**异常消息**/
    private final String X_EXCEPTION_MESSAGE = "x-exception-message";
    /**异常栈**/
    private final String X_EXCEPTION_STACKTRACE = "x-exception-stacktrace";
    /**交换机**/
    private final String X_ORIGINAL_EXCHANGE = "x-original-exchange";
    /**路右键**/
    private final String X_ORIGINAL_ROUTINGKEY = "x-original-routingKey";
    /**初始化重试次数**/
    private final Integer INIT_RETRY_NUM = 0;


    @Autowired
    private MongoTemplate mongoTemplate;
    @Autowired
    private TomatoMsProperties tomatoMsProperties;

    @Override
//    @ReliabilityStrategy
    public void process(Message message, Channel channel) {

        try {
            //msgId
            String id = message.getMessageProperties().getMessageId();
            RetryMessage retryMessage = null;
            //没有id
            if(Objects.isNull(id)){
                retryMessage = new RetryMessage();
                doProcessNewMessage(message,retryMessage);
                return;
            }

            //有id但是查询不到

            retryMessage = findByMsgId(id);

            if(Objects.isNull(retryMessage)){
                retryMessage = new RetryMessage();
                doProcessNewMessage(message,retryMessage);
                return;
            }

            //有id并且查到了数据
            doProcessOldMessage(retryMessage);
        } catch (Exception e) {
            log.error("【处理死信队列异常,异常信息为:{},消息为:{}】",e.getMessage(),message);
        }
    }

    private RetryMessage findByMsgId(String id) {
        Query query = new Query();
        query.addCriteria(Criteria.where(MSG_ID).is(id));
        List<RetryMessage> retryMessages = mongoTemplate.find(query, RetryMessage.class);
        if(CollectionUtils.isEmpty(retryMessages)){
            return null;
        }
        return retryMessages.get(0);
    }

    /**
     * @Author huqiang
     * @Description //TODO 处理旧的消息（即失败过的消息)
     * @Date  2019-11-23
     * @Param message
     * @Param retryMessage
     * @return void
     */
    private void doProcessOldMessage(RetryMessage retryMessage) {
        Integer maxRetryNum = tomatoMsProperties.getMaxRetryNum();
        int retryNum = retryMessage.getRetryNum();
        ++retryNum;
        String id = retryMessage.getId();
        Query query = new Query();
        Criteria criteria = Criteria.where(ID).is(id);
        Update update = new Update();
        update.set(RETRY_NUM,retryNum);
        //超过最大重试次数 则更新字段标识不能再被重发
        if(retryNum==maxRetryNum){
            update.set(CAN_CONTINUE,Boolean.FALSE);
        }
        update.set(UPDATE_TIME,LocalDateTime.now());
        query.addCriteria(criteria);
        UpdateResult updateResult = mongoTemplate.updateFirst(query, update, RetryMessage.class);
        long matchedCount = updateResult.getMatchedCount();
        if(matchedCount<1){
            throw new RuntimeException("未更新到具体的值,更新id为:"+id+",更新的消息重试次数为:"+retryNum);
        }

    }


    /**
     * @Author huqiang
     * @Description //TODO 处理新的消息（即第一次失败的消息)
     * @Date  2019-11-23
     * @Param message
     * @Param retryMessage
     * @return void
     */
    private void doProcessNewMessage(Message message,RetryMessage retryMessage) {
        LocalDateTime localDateTime = LocalDateTime.now();

        byte[] body = message.getBody();
        MessageProperties messageProperties = message.getMessageProperties();
        Map<String, Object> headers = messageProperties.getHeaders();
        headers.remove(X_EXCEPTION_STACKTRACE);
        String headsJson = JacksonUtil.bean2JsonByZoneTime(headers);
        retryMessage.setBody(new String(body, StandardCharsets.UTF_8));
        retryMessage.setHeaders(headsJson);

        if(!CollectionUtils.isEmpty(headers)){
            Object exceptionMsg = headers.get(X_EXCEPTION_MESSAGE);
            Object routingKey = headers.get(X_ORIGINAL_ROUTINGKEY);
            Object exchange = headers.get(X_ORIGINAL_EXCHANGE);


            Optional.ofNullable(exceptionMsg).ifPresent(e->retryMessage.setExceptionMsg(e.toString()));

            Optional.ofNullable(routingKey).ifPresent(e->retryMessage.setRoutingKey(e.toString()));

            Optional.ofNullable(exchange).ifPresent(e->retryMessage.setExchange(e.toString()));

            retryMessage.setRetryNum(INIT_RETRY_NUM);

            retryMessage.setInputTime(localDateTime);

            retryMessage.setUpdateTime(localDateTime);

            retryMessage.setCanContinue(Boolean.TRUE);
            String messageId = messageProperties.getMessageId();
            retryMessage.setMsgId(messageId);
        }
        mongoTemplate.save(retryMessage);
    }
}
